package com.allyes.flume.interceptor.parser;

import com.allyes.flume.interceptor.DynamicRawlogEventHeader;
import com.allyes.flume.interceptor.InvalidLogHandler;
import com.allyes.flume.interceptor.rawlogdigger.DiggerInfoFetcher;
import com.allyes.flume.interceptor.parser.ParserException.*;
import com.allyes.flume.source.Pair;
import com.allyes.log.LogType;
import com.allyes.log.RawLog;

import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.Encoder;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.flume.Event;
import org.apache.flume.event.EventBuilder;

import java.io.ByteArrayOutputStream;
import java.io.IOException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * Created by gc on 7/14/16.
 */
public class LogParserAinsight extends LogParser {
  private static final Logger logger = LoggerFactory.getLogger(LogParserAinsight.class);


  public LogParserAinsight() {
    invalidLogHandler = null;
    dynamicRawlogEventHeader = null;
  }

  public LogParserAinsight(InvalidLogHandler h, DynamicRawlogEventHeader dh) {
    invalidLogHandler = h;
    dynamicRawlogEventHeader = dh;
  }

  @Override
  public Pair<Boolean, Event> parseToEvent(
      long inode,
      String fileName,
      int lineIndex,
      String line) throws IOException {
    Pair<Boolean, Event> ret = new Pair<Boolean, Event>();

    try {
      RawLog rawlog = toRawlog(line, fileName, lineIndex);
      byte[] serializedRawLog = converToBinary(rawlog);
      Event event = EventBuilder.withBody(serializedRawLog);

      if (dynamicRawlogEventHeader != null) {
        dynamicRawlogEventHeader.updateHeader(event, new DiggerInfoFetcher(rawlog));
      }

      ret.left = true;
      ret.right = event;
      return ret;

    } catch (InvalidLogException e) {
      ret.left = false;
      if (invalidLogHandler != null) {
        ret.right = invalidLogHandler.handle(inode, fileName, lineIndex, line, e.getMessage());
      } else {
        ret.right = null;
      }
      logger.error(e + " in (file=" + fileName + ", inode=" + inode + ", lineIndex=" + lineIndex + ")");
      return ret;
    }
  }

  public com.allyes.log.RawLog toRawlog(String line, String filename, int lineno)
      throws LogFormatException, LogTypeException {
    RawLog rawLog = new RawLog();

    String[] ss =  line.split("\t", -1);
    if (ss.length >= 20) {
      int pos = 0;
      rawLog.setTimestamp(getTimestamp(ss[pos++]));
      rawLog.setTimestampMs(getTimestampMS(ss[pos++]));
      rawLog.setLogType(getLogType(ss[pos++]));
      rawLog.setProductName(ss[pos++]);
      rawLog.setDb(ss[pos++]);
      pos++;  // interval (for debug)
      rawLog.setIsNewUser(isNewUser(ss[pos++]));
      rawLog.setAllyesid(ss[pos++]);
      rawLog.setThirdPartyID(ss[pos++]);
      rawLog.setIp(ss[pos++]);
      rawLog.setClientIp(ss[pos++]);
      rawLog.setForwardedIp(ss[pos++]);
      rawLog.setRequestUrl(ss[pos++]);
      rawLog.setHttpReferer(ss[pos++]);
      rawLog.setUserAgent(ss[pos++]);
      rawLog.setLanguage(ss[pos++]);
      rawLog.setDeviceId(ss[pos++]);
      rawLog.setMobileNum(ss[pos++]);
      rawLog.setCheatingIndex(getCheatingIndex(ss[pos++]));
      rawLog.setHttpProtocolHeader(ss[pos++]);
      rawLog.setFileName(filename);
      rawLog.setLineNo(lineno);
    } else {
      throw new LogFormatException("log-fileds-not-enough-only-" + ss.length);
    }

    return rawLog;
  }

  public static byte[] converToBinary(RawLog rawLog) throws IOException {
    DatumWriter<RawLog> datumWriter = new SpecificDatumWriter<RawLog>(RawLog.class);
    ByteArrayOutputStream out = new ByteArrayOutputStream();
    Encoder encoder = EncoderFactory.get().binaryEncoder(out, null);
    datumWriter.write(rawLog, encoder);
    encoder.flush();
    out.close();
    return out.toByteArray();
  }

  private long getTimestamp(String input) throws LogFormatException {
    long ret = 0;
    try {
      ret = Long.parseLong(input);
    } catch (NumberFormatException e) {
      throw new LogFormatException("timestamp-failed-to-parseLong-" + input);
    }

    return ret;
  }

  private int getTimestampMS(String input) throws LogFormatException {
    int ret = 0;
    try {
      ret = Integer.parseInt(input);
    } catch (NumberFormatException e) {
      throw new LogFormatException("timestamp-ms-failed-to-parseInt-" + input);
    }

    return ret;
  }

  private LogType getLogType(String type) throws LogTypeException {
    int iType = 0;
    try {
      iType = Integer.parseInt(type);
    } catch (NumberFormatException e) {
      throw new ParserException.LogTypeException("log-type-failed-to-parseInt-" + type);
    }

    LogType logType = null;
    switch(iType) {
      case 1:
        logType = LogType.BID;
        break;
      case 2:
        logType = LogType.SHOW;
        break;
      case 3:
        logType = LogType.NOBID;
        break;
      case 4:
        logType = LogType.CLICK;
        break;
      case 5:
        logType = LogType.TRACK;
        break;
      default:
        throw new LogTypeException("log-type-not-found-" + type);
    }

    return logType;
  }

  private boolean isNewUser(String newUser) throws LogFormatException {
    int i = 0;
    try {
      i = Integer.parseInt(newUser);
    } catch (NumberFormatException e) {
      throw new LogFormatException("new_user-failed-to-parseInt-" + newUser);
    }

    switch (i) {
      case 1: return true;
      case 0: return false;
      default: throw new LogFormatException("new_user-invalid-value-" + newUser);
    }
  }

  private int getCheatingIndex(String input) throws LogFormatException {
    int ret = 0;
    try {
      ret = Integer.parseInt(input);
    } catch (NumberFormatException e) {
      throw new LogFormatException("cheating_index-failed-to-parseInt-" + input);
    }

    return ret;
  }
}
